import { RedisService } from './redis.service';
import { Injectable, Inject, forwardRef } from '@nestjs/common';
import * as _ from 'lodash';
import { globalError } from '../interceptors/httpResult.interceptor';

interface ResolveRejectStorage {
    keys: string[];
    resolve: (data: Unlocker) => void;
    reject: (err: any) => void;
}

class Unlocker {
    private lock: LockService = null;
    private keys: string[] = null;

    constructor(lock: LockService, keys: string[]) {
        this.lock = lock;
        this.keys = keys;
    }

    async release() {
        if (this.keys == null) return;
        let keys = this.keys;
        this.keys = null;
        await this.lock.releaseKeys(keys);
    }
}

@Injectable()
export class LockService {
    private redisChannelName = 'redis-lock';

    constructor(
        @Inject(forwardRef(() => RedisService))
        private redisService: RedisService,
    ) {
        let client = this.redisService.subscribeRedisClient();
        console.log(client)
        client.subscribe(this.redisChannelName);
        client.on(this.redisChannelName, () => {
            this.lockReally();
        });
    }

    private waiting: ResolveRejectStorage[] = [];

    private transferLockKeys(keys: string[]) {
        return keys.map(key => `redis-lock-key:${key}`);
    }

    private transferMset(keys: string[]) {
        return _.flatten(keys.map(key => [key, 1]));
    }

    /**
     * 使用 msetnx 命令来控制锁
     */
    private async runningIncludeKeys(keys: string[]): Promise<boolean> {
        let lockKeys = this.transferLockKeys(keys);
        let success = await this.redisService
            .dataRedisClient()
            .msetnx(...this.transferMset(lockKeys));
        return success === 1;
    }

    /**
     * 广播可以试着开启新任务
     */
    private broadcastTryNewTask() {
        this.redisService
            .publishRedisClient()
            .publish(this.redisChannelName, '');
    }

    async runTransaction<T = any>(
        key: string | string[],
        fn: () => T | Promise<T>,
    ): Promise<T> {
        const lock = await this.lock(key);
        try {
            const result = fn();
            return result instanceof Promise ? await result : result;
        } catch (e) {
            console.error(e);
            if (e instanceof Error) {
                e = e.message;
            }
            globalError(e);
        } finally {
            await lock.release();
        }
    }

    lock(key: string | string[]): Promise<Unlocker> {
        let keys: string[];
        if (Array.isArray(key)) {
            keys = key;
        } else {
            keys = [key];
        }
        return new Promise<Unlocker>((resolve, reject) => {
            this.waiting.push({
                resolve,
                reject,
                keys,
            });
            this.lockReally();
        });
    }

    async releaseKeys(keys: string[]) {
        await this.redisService
            .dataRedisClient()
            .del(...this.transferLockKeys(keys));
        this.broadcastTryNewTask();
        this.lockReally();
    }

    private async lockReally() {
        for (let task of this.waiting) {
            let canTaskRun = await this.runningIncludeKeys(task.keys);
            if (canTaskRun) {
                // 在waiting中remove掉
                let index = this.waiting.indexOf(task);
                this.waiting.splice(index, 1);

                let unlock = new Unlocker(this, task.keys);
                task.resolve(unlock);

                // 如果10s之后还没有release自动unlock
                setTimeout(() => {
                    unlock.release();
                }, 10 * 1000);
            }
        }
    }
}
